#include <stdlib.h>
#include <string.h>
#include <stdio.h>

#include "list.h"
#include "server_type.h"
#include "socketlibfunction.h"
#include "mutex.h"
#include "server.h"

static void buffer_delete(struct buffer_s* self)
{
    if(NULL != self)
    {
        if(NULL != self->data)
        {
            free(self->data);
            self->data = NULL;
        }

        free(self);
        self = NULL;
    }
}

static struct buffer_s* buffer_new(int buffer_size)
{
    struct buffer_s* ret = (struct buffer_s*)malloc(sizeof(struct buffer_s));
    
    if(NULL != ret)
    {
        memset(ret, 0, sizeof(*ret));

        if(NULL != (ret->data = (char*)malloc(sizeof(char)*buffer_size)))
        {
            memset(ret->data, 0, sizeof(char)*buffer_size);
            ret->data_len = buffer_size;
            ret->read_pos = 0;
            ret->write_pos = 0;
        }
        else
        {
            buffer_delete(ret);
            ret = NULL;
        }
    }

    return ret;
}

void buffer_adjustto_head(struct buffer_s* self)
{
    int len = 0;

    if(self->read_pos > 0)
    {
        len = self->write_pos - self->read_pos;
        if(len > 0 )
        {
            memmove(self->data, self->data+self->read_pos, len);
        }
        self->read_pos = 0;
        self->write_pos = len;
    }
}

int session_recv(struct session_s* self)
{
    int recv_result = RECV_BUFFERSMALL;
    int can_recv_size = 0;
    int recv_buffer_len = self->recv_buffer->data_len;
    int recv_len = 0;
    int fd = self->fd;
    struct buffer_s* buffer = self->recv_buffer;
    char* buffer_data = buffer->data;

    if(NULL != buffer_data)
    {
        if(buffer->write_pos == buffer->data_len)
        {
            buffer_adjustto_head(buffer);
        }

        can_recv_size =  recv_buffer_len - buffer->write_pos;
        if(can_recv_size > 0)
        {
            recv_len = recv(fd, buffer_data+buffer->write_pos, can_recv_size, 0);
            if(0 == recv_len)
            {
                recv_result |= RECV_CLOSE;
            }
            else if(recv_len == SOCKET_ERROR)
            {
                if(S_EWOULDBLOCK == sErrno)
                {
                    recv_result |= RECV_EAGIN;
                }
                else
                {
                    recv_result |= RECV_CLOSE;
                }
            }
            else
            {
                buffer->write_pos += recv_len;
                if(recv_len < can_recv_size)
                {
                  recv_result |= RECV_EAGIN;
                }
                
                recv_result |= RECV_OK;
            }
        }
    }
    
    return recv_result;
}

int session_send(struct session_s* self)
{
    int send_result = SEND_OK;
    int transnum = 0;
    struct buffer_s* buffer = self->send_buffer;
    int send_size = buffer->write_pos-buffer->read_pos;

    if(send_size > 0)
    {
        mutex_lock(self->send_mutex);
        transnum = send(self->fd, buffer->data+buffer->read_pos, send_size, 0);
        if(transnum > 0)
        {
            buffer->read_pos+=transnum;
            if(buffer->write_pos == buffer->read_pos)
            {
                buffer_adjustto_head(self->send_buffer);
            }

            if(transnum < send_size)
            {
                send_result |= SEND_EAGIN;
            }
        }
        else if(SOCKET_ERROR == transnum)
        {
            send_result |= (S_EWOULDBLOCK == sErrno ? SEND_EAGIN : SEND_CLOSE);
        }

        mutex_unlock(self->send_mutex);
    }

    return send_result;
}

static void session_destroy(struct session_s* self)
{
    if(NULL != self->send_buffer)
    {
        buffer_delete(self->send_buffer);
        self->send_buffer = NULL;
    }
    
    if(NULL != self->recv_buffer)
    {
        buffer_delete(self->recv_buffer);
        self->recv_buffer = NULL;
    }
    
    if(NULL != self->send_mutex)
    {
        mutex_delete(self->send_mutex);
        self->send_mutex = NULL;
    }

    if(NULL != self->ext_data)
    {
        free(self->ext_data);
        self->ext_data = NULL;
    }
    
}

static bool session_construction(struct session_s* self, int index, int send_buffer_size, int recv_buffer_size)
{
    if(NULL != (self->send_buffer = buffer_new(send_buffer_size)) &&
       NULL != (self->recv_buffer = buffer_new(recv_buffer_size)) &&
       NULL != (self->send_mutex = mutex_new()))
    {
        self->index = index;
        self->fd = SOCKET_ERROR;
        self->readable = false;
        self->writable = false;
        self->status = SOCKET_FREE;
        self->ext_data = NULL;

        return true;
    }
    else
    {
        session_destroy(self);
        return false;
    }
}

bool server_construction(struct server_s* self, int num, int recv_buffer_size, int send_buffer_size)
{
    memset(self, 0, sizeof(*self));

    if(NULL != (self->wait_close_list = rwlist_new(num, sizeof(int))) &&
       NULL != (self->accepted_list = rwlist_new(num, sizeof(sock))) &&
       NULL != (self->readable_list = list_new(num, sizeof(struct session_s*))) &&
       NULL != (self->writable_list = list_new(num, sizeof(struct session_s*))) &&
       NULL != (self->free_sockets = stack_new(num, sizeof(sizeof(struct session_s*)))) &&
       NULL != (self->sockets = array_new(num, sizeof(struct session_s))))
    {
        int i = 0;
        self->socket_num = num;
        self->ext = NULL;

        for(; i < num; ++i)
        {
            struct session_s* client = (struct session_s*)array_at(self->sockets, i);
            if(!session_construction(client, i, send_buffer_size, recv_buffer_size))
            {
                break;
            }

            stack_push(self->free_sockets, &client);
        }

        if(i < num)
        {
            server_destroy(self);
            return false;
        }
    }

    return true;
}

void server_destroy(struct server_s* self)
{
    if(NULL != self->wait_close_list)
    {
        rwlist_delete(self->wait_close_list);
        self->wait_close_list = NULL;
    }
    
    if(NULL != self->accepted_list)
    {
        rwlist_delete(self->accepted_list);
        self->accepted_list = NULL;
    }
    
    if(NULL != self->readable_list)
    {
        list_delete(self->readable_list);
        self->readable_list = NULL;
    }
    
    if(NULL != self->writable_list)
    {
        list_delete(self->writable_list);
        self->writable_list = NULL;
    }

    if(NULL != self->sockets)
    {
        struct array_s* sockets = self->sockets;
        int num = self->socket_num;
        int i = 0;

        for(; i < num; ++i)
        {
            struct session_s* client = (struct session_s*)array_at(sockets, i);
            session_destroy(client);
        }

        array_delete(sockets);
        self->sockets = NULL;
    }
    
    if(NULL != self->free_sockets)
    {
        stack_delete(self->free_sockets);
        self->free_sockets = NULL;
    }
}

void server_enable_session_readable(struct server_s* self, struct session_s* session)
{
    if(!session->readable)
    {
        list_push_back(self->readable_list, &session);
        session->readable = true;
    }
}

void server_enable_session_writable(struct server_s* self, struct session_s* session)
{
    if(!session->writable)
    {
        list_push_back(self->writable_list, &session);
        session->writable = true;
    }
}

void server_disable_session_readable(struct server_s* self, struct session_s* session)
{
    session->readable = false;
}

void server_disable_session_writable(struct server_s* self, struct session_s* session)
{
    session->writable = false;
}

void server_remove_readable(struct server_s* self, struct session_s* session)
{
    struct list_s* temp_list = self->readable_list;
    struct list_node_s* node = list_begin(temp_list);
    const struct list_node_s* tail = list_end(temp_list);

    if(session->readable)
    {
        for(; node != tail; )
        {
            if(session == *(struct session_s**)(node->data))
            {
                node = list_erase(temp_list, node);
                break;
            }
            
            node = node->next;
        }
    }
}

void server_remove_writable(struct server_s* self, struct session_s* session)
{
    struct list_s* temp_list = self->writable_list;
    struct list_node_s* node = list_begin(temp_list);
    const struct list_node_s* tail = list_end(temp_list);

    if(session->writable)
    {
        for(; node != tail; )
        {
            if(session == *(struct session_s**)(node->data))
            {
                node = list_erase(temp_list, node);
                break;
            }

            node = node->next;
        }
    }
}

static void session_init(struct session_s* self, sock fd)
{
    self->fd = fd;
    self->status = SOCKET_CONNECTED;
    socket_nonblock(fd);
}

void session_close(struct session_s* self)
{
    self->readable = false;
    self->writable = false;
    self->recv_buffer->read_pos = 0;
    self->recv_buffer->write_pos = 0;
    self->send_buffer->read_pos = 0;
    self->send_buffer->write_pos = 0;
    memset(self->recv_buffer->data, 0, self->recv_buffer->data_len);
    self->status = SOCKET_FREE;
    self->fd = SOCKET_ERROR;
}

static void server_handler_accepted(struct server_s* self)
{
    struct stack_s* accept_list = rwlist_get_readlist(self->accepted_list);
    sock fd = SOCKET_ERROR;
    struct session_s** client_pp = NULL;

    while(true)
    {
        void* fd_ptr = stack_pop(accept_list);
        if(NULL == fd_ptr)
        {
            break;
        }

        fd = *(sock*)fd_ptr;
        client_pp = (struct session_s**)stack_pop(self->free_sockets);
        if(NULL != client_pp)
        {
            struct session_s* client = *client_pp;
            session_init(client, fd);
            (self->enter)(self, client);
            (self->logic_on_enter)(self, client->index);
        }
        else
        {
            printf("has no session give new client\n");
            socket_close(fd);
        }
    }
}

static void server_handler_waitclose(struct server_s* self)
{
    if(rwlist_get_readlistnum(self->wait_close_list) > 0)
    {
        struct stack_s* wait_close_list = rwlist_get_readlist(self->wait_close_list);
        int index = 0;
        struct session_s* client = NULL;

        while(true)
        {
            void* index_ptr = stack_pop(wait_close_list);
            if(NULL == index_ptr)
            {
                break;
            }

            index = *(int*)index_ptr;
            client = (struct session_s*)array_at(self->sockets, *(int*)index_ptr);

            if(NULL != client && client->status != SOCKET_FREE)
            {
                self->close(self, client);
            }
        }
    }
}

void server_poll(struct server_s* self)
{
    server_handler_waitclose(self);
    self->poll(self);
    self->handler_io(self);
}

void server_socket_add(struct server_s* self, sock fd)
{
    rwlist_push(self->accepted_list, &fd);
}

void server_socket_close(struct server_s* self, int index)
{
    rwlist_push(self->wait_close_list, &index);
}

int server_socket_send(struct server_s* self, int index, const char* data, int len)
{
    int can_send_len = 0;
    int send_len = 0;
    struct buffer_s* buffer = NULL;
    char* new_data = NULL;
    int new_len = 0;
    struct session_s* client = (struct session_s*)array_at(self->sockets, index);

    if(NULL != client)
    {
        mutex_lock(client->send_mutex);
        buffer = client->send_buffer;
        
        do 
        {
            can_send_len  = buffer->data_len-buffer->write_pos;
            if(can_send_len < len)
            {
                new_len = (sizeof(char)*buffer->data_len*2+len);
                new_data = (char*)malloc(new_len);
                if(NULL != new_data)
                {
                    buffer_adjustto_head(buffer);
                    memcpy(new_data, buffer->data, buffer->write_pos-buffer->read_pos);
                    free(buffer->data);
                    buffer->data = new_data;
                    buffer->data_len = new_len;
                }
                else
                {
                    break;
                }
            }
            
            memcpy(buffer->data+buffer->write_pos, data, len);
            buffer->write_pos += len;
            send_len = len;
        }while(0);

        mutex_unlock(client->send_mutex);

        if(NULL != self->socket_on_logic_send)
        {
            self->socket_on_logic_send(self, client);
        }
    }

    return send_len;
}

void server_handler_recv(struct server_s* self)
{
    struct list_s* temp_list = self->readable_list;
    struct list_node_s* node = list_begin(temp_list);
    const struct list_node_s* tail = list_end(temp_list);
    server_recv_pt  temp_recv = self->recv;

    for(; node != tail; )
    {
        struct session_s* session = *(struct session_s**)(node->data);
        node = node->next;

        if(!session->readable || temp_recv(self, session))
        {
            session->readable = false;
            node = list_erase(temp_list, node->prior);
        }
    }
}

void server_handler_send(struct server_s* self)
{
    struct list_s* temp_list = self->writable_list;
    struct list_node_s* node = list_begin(temp_list);
    const struct list_node_s* tail = list_end(temp_list);
    server_send_pt temp_send = self->send;

    for(; node != tail; )
    {
        struct session_s* session = *(struct session_s**)(node->data);
        node = node->next;
        
        if(!session->writable || temp_send(self, session))
        {
            session->writable = false;
            node = list_erase(temp_list, node->prior);
        }
    }
}

void server_setext(struct server_s* self, void* ext)
{
    self->ext = ext;
}

void* server_getext(struct server_s* self)
{
    return self->ext;
}

void server_socket_onclose(struct server_s* self, struct session_s* session)
{
    session_close(session);
    stack_push(self->free_sockets, &session);
}